package com.leyantech.ecp.agenx.session.manager;

import com.leyantech.chaos.kafka.KafkaConsumerStringClient;
import com.leyantech.chaos.kafka.KafkaProducerStringClient;
import com.leyantech.ecp.agenx.common.ClusterVerticle;
import com.leyantech.ecp.agenx.session.manager.di.SessionConsumerModule;
import com.leyantech.ecp.agenx.session.manager.handler.AssistantLogoutHandler;

import com.google.common.collect.Lists;
import com.google.inject.Inject;
import com.google.inject.Module;
import lombok.SneakyThrows;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;

import java.util.List;

/**
 * @author eahau, {@literal <eahau@foxmail.com>}
 * @date 2022-11-07.
 */
public class SessionConsumerVerticle extends ClusterVerticle {

  @Inject
  private KafkaConsumerStringClient consumer;

  @Inject
  private KafkaProducerStringClient producer;

  @Inject
  private PathChildrenCache clusterNodeInfo;

  @Inject
  private AssistantLogoutHandler logoutHandler;

  @Override
  protected List<Module> dependentModules() {
    return Lists.newArrayList(new SessionConsumerModule());
  }

  private void startHealthCheckServer() {
    vertx.createHttpServer()
        .requestHandler(request -> request.response().send("ok"))
        .listen(8080);
  }

  @SneakyThrows
  @Override
  public void start() {
    clusterNodeInfo.getListenable().addListener(logoutHandler);
    consumer.start();
    startHealthCheckServer();
  }

  @Override
  public void stop() {
    consumer.destroy();
    producer.destroy();
  }

  public static void main(String[] args) {
    new SessionConsumerVerticle().startVerticle();
  }

}
